import com.totoro.rpc.common.bean.RPCRequest;
import com.totoro.rpc.common.bean.RPCResponse;
import com.totoro.rpc.common.codec.RPCDecoder;
import com.totoro.rpc.common.codec.RPCEncoder;
import com.totoro.rpc.common.exception.RPCServiceException;
import com.totoro.rpc.common.util.NetworkUtil;
import com.totoro.rpc.common.util.StringUtils;
import com.totoro.rpc.register.Discover;
import com.totoro.rpc.register.Register;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import lombok.NonNull;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.net.InetSocketAddress;

/**
 * 客户端服务调用
 * Created by Niki on 2018/5/23 10:59
 */
@Slf4j
@RequiredArgsConstructor()
public class RPCClient extends SimpleChannelInboundHandler<RPCResponse> {
    @NonNull
    private Discover discover;
    private Register register;
    private RPCResponse response;

    protected RPCResponse send(RPCRequest request) throws RPCServiceException, InterruptedException {
        InetSocketAddress inetSocketAddress = discover(request);
        EventLoopGroup group = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(group);
        try {
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    socketChannel.pipeline().addLast(new RPCEncoder(RPCRequest.class));
                    socketChannel.pipeline().addLast(new RPCDecoder(RPCResponse.class));
                    socketChannel.pipeline().addLast(RPCClient.this);
                }
            });
            bootstrap.option(ChannelOption.TCP_NODELAY, true);

            //连接RPC服务器
            ChannelFuture future = bootstrap.connect(inetSocketAddress).sync();

            //写RPC请求
            Channel channel = future.channel();
            channel.writeAndFlush(request).sync();

            channel.closeFuture().sync();

            return this.response;
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * 获取服务提供方地址
     *
     * @param request
     * @return
     * @throws RPCServiceException
     */
    private InetSocketAddress discover(RPCRequest request) throws RPCServiceException {
        if (request == null) {
            throw new RPCServiceException("请求对象为空！");
        }
        String serviceName = request.getServiceName();
        String version = request.getVersion();
        if (StringUtils.isBlank(serviceName) || StringUtils.isBlank(version)) {
            throw new RPCServiceException("请求服务的服务名以及版本不能为空！");
        }

        String address = discover.discover(serviceName, version);
        if (StringUtils.isBlank(address)) {
            throw new RPCServiceException("无法获取版本为：" + version + "的服务：" + serviceName);
        }

        NetworkUtil.serverHeath(address);
        String[] args = address.split(":");
        int port = Integer.parseInt(args[1]);
        return new InetSocketAddress(args[0], port);
    }

    /**
     * 想注册中心注册该消费者，在启动时注册一次
     */
    private void registerClient() {
        register.registerClient();
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        log.error("调用服务异常！", cause);
        ctx.close();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, RPCResponse rpcResponse) throws Exception {
        this.response = rpcResponse;
    }

}
